package cn.linjianhui.demo.flink;

import org.apache.flink.api.common.functions.FoldFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
import org.apache.flink.util.Collector;

public class ConnectedStreamsTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStream<Long> source = env.generateSequence(1, 5);

        DataStream<Long> target = env.generateSequence(6, 10);

        ConnectedStreams<Long, Long> connectedStreams = source.connect(target.broadcast());

        DataStream<String> result =
            connectedStreams.flatMap(new CoFlatMapFunction<Long, Long, String>(){
                @Override
                public void flatMap1(Long a, Collector<String> out) throws Exception {
                    out.collect(a.toString());
                }

                @Override
                public void flatMap2(Long b, Collector<String> out) throws Exception {
                    out.collect(b.toString());
                }
            });

        result.print();
        env.execute();
    }
}
